package org.chenwei;

import java.io.IOException;
import java.util.Iterator;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.*;

/**
 * This class implements the map reduce job, 
 * which gets the minimum exchange rate from
 * USD to JPY in each minute. 
 *
 * @author chenwei
 */
public class MinExchangeRate
{
    /**
     * This inner class is a Mapper to map the input 
     * data to <key1, value> pairs.
     * 
     * The input pair type is <Long, Text> and the 
     * output pair type is <Text, Double>.
     */
    public static class MinExchangeRateMapper extends MapReduceBase
        implements Mapper<LongWritable, Text, Text, DoubleWritable>
    {
        /**
         * This method is the core map method, which 
         * map the input data described as .cvs records
         * to <key1, value> pairs for Reducer.
         * @param key1
         * @param value
         * @param output
         * @param reporter
         * @throws IOException 
         */
        @Override
        public void map(LongWritable key1, Text value,
            OutputCollector<Text, DoubleWritable> output,
            Reporter reporter) throws IOException
        {
            String line = value.toString();
            String[] items = line.split(",");
            
            String datetime = items[2].substring(1, items[2].length() - 1) 
                + " " + items[3].substring(1, items[3].length() - 1);
            double rate = Double.parseDouble(items[1]);
            
            output.collect(new Text(datetime), new DoubleWritable(rate));
        }
    }

    /**
     * This inner class is a Reducer to reduce the 
     * <key2, values> from Mapper to find out the 
     * minimum exchange rate in each minute. 
     * 
     * The input pair type is <Text, Double> and the 
     * output pair type is <Text, Double>.
     */
    public static class MinExchangeRateReducer extends MapReduceBase
        implements Reducer<Text, DoubleWritable, Text, DoubleWritable>
    {
        /**
         * This method tries to find out the minimum 
         * exchange rate in each minute.
         * @param key2
         * @param values
         * @param output
         * @param reporter
         * @throws IOException 
         */
        @Override
        public void reduce(Text key2, Iterator<DoubleWritable> values,
            OutputCollector<Text, DoubleWritable> output,
            Reporter reporter) throws IOException
        {
            double minValue = Double.POSITIVE_INFINITY;
            while (values.hasNext()) {                
                minValue = Math.min(minValue, values.next().get());
            }
            
            output.collect(key2, new DoubleWritable(minValue));
        }
    }

    /**
     * This method is the main entry to the program, 
     * which first configures the job to run on Hadoop, 
     * and then runs the job.
     * @param args the command line arguments
     */
    public static void main(String[] args) throws IOException
    {
        JobConf conf = new JobConf(MinExchangeRate.class);
        conf.setJobName("Min exchange rate");
        
        conf.setOutputKeyClass(Text.class);
        conf.setOutputValueClass(DoubleWritable.class);
        
        conf.setMapperClass(MinExchangeRateMapper.class);
        conf.setCombinerClass(MinExchangeRateReducer.class);
        conf.setReducerClass(MinExchangeRateReducer.class);
        
        conf.setInputFormat(TextInputFormat.class);
        conf.setOutputFormat(TextOutputFormat.class);
        
        FileInputFormat.setInputPaths(conf, new Path(args[0]));
        FileOutputFormat.setOutputPath(conf, new Path(args[1]));
        
        JobClient.runJob(conf);
    }
}
